package org.nobody.mallb2c.mallseckill.service.impl;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.nobody.common.to.mq.SeckillOrderTo;
import org.nobody.common.utils.R;
import org.nobody.common.vo.MemberRespVo;
import org.nobody.mallb2c.mallseckill.feign.CouponFeignService;
import org.nobody.mallb2c.mallseckill.feign.ProductFeignService;
import org.nobody.mallb2c.mallseckill.interceptor.LoginUserInterceptor;
import org.nobody.mallb2c.mallseckill.service.SeckillService;
import org.nobody.mallb2c.mallseckill.to.SeckillSkuRedisTo;
import org.nobody.mallb2c.mallseckill.vo.SeckillSessionsWithSkus;
import org.nobody.mallb2c.mallseckill.vo.SeckillSkuVo;
import org.nobody.mallb2c.mallseckill.vo.SkuInfoVo;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author 3489309323@qq.com
 * @create 2020-09-28-13:50
 */
@Slf4j
@Service
public class SeckillServiceImpl implements SeckillService {
  @Autowired
  CouponFeignService couponFeignService;

  @Autowired
  StringRedisTemplate redisTemplate;

  @Autowired
  ProductFeignService productFeignService;

  @Autowired
  RedissonClient redissonClient;

  @Autowired
  RabbitTemplate rabbitTemplate;

  private final String SESSIONS_CACHE_PREFIX = "seckill:sessions:";
  private final String SKUKILL_CACHE_PREFIX = "seckill:skus:";
  private final String SKU_STOCK_SEMAPHORE = "seckill:stock:"; //+商品随机码


  /** 自定义受sentinel保护的资源 方式②
   * 注解的方式
   *
   * 注意 blockHandler 函数会在原方法被限流/降级/系统保护的时候调用，
   * 而 fallback 函数会针对所有类型的异常。
   * 请注意 blockHandler 和 fallback 函数的形式要求，
   * 更多指引可以参见 Sentinel 注解支持文档。*/
  /*public List<SeckillSkuRedisTo> blockHandler(BlockException e){
    log.error("getCurrentSeckillSkusResource被限流了");
    return null;
  }
  @SentinelResource(value = "getCurrentSeckillSkusResource", blockHandler = "blockHandler")*/
  @Override
  public List<SeckillSkuRedisTo> getCurrentSeckillSkus() {
    //1、确定当前时间属于哪个秒杀场次
    long time = new Date().getTime();

    /** 自定义受sentinel保护的资源 方式①
     * try catch的方式*/
    /*try(Entry entry = SphU.entry("seckillSkus")) {
      Set<String> keys = redisTemplate.keys(SESSIONS_CACHE_PREFIX + "*");
      for (String key : keys) {
        //seckill:sessions:1601294400000_1601298000000
        String replace = key.replace(SESSIONS_CACHE_PREFIX, "");
        String[] s = replace.split("_");
        Long start = Long.parseLong(s[0]);
        Long end = Long.parseLong(s[1]);
        if (time >= start && time <= end) {
          //2、获取这个秒杀场次需要的所有商品信息
          List<String> range = redisTemplate.opsForList().range(key, -100, 100);//emmm 如果超过100件要修改一下
          BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
          List<String> list = hashOps.multiGet(range);
          if (list != null) {
            List<SeckillSkuRedisTo> collect = list.stream().map(item -> {
              SeckillSkuRedisTo redisTo = JSON.parseObject((String) item, SeckillSkuRedisTo.class);
              //redisTo.setRandomCode(null); //当前秒杀开始了 需要随机码
              return redisTo;
            }).collect(Collectors.toList());
            return collect;
          }
          break;
        }
      }
    }catch (BlockException e){
      log.error("资源被限流: {}",e.getMessage());
    }*/

    Set<String> keys = redisTemplate.keys(SESSIONS_CACHE_PREFIX + "*");
    for (String key : keys) {
      //seckill:sessions:1601294400000_1601298000000
      String replace = key.replace(SESSIONS_CACHE_PREFIX, "");
      String[] s = replace.split("_");
      Long start = Long.parseLong(s[0]);
      Long end = Long.parseLong(s[1]);
      if (time >= start && time <= end) {
        //2、获取这个秒杀场次需要的所有商品信息
        List<String> range = redisTemplate.opsForList().range(key, -100, 100);//emmm 如果超过100件要修改一下
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
        List<String> list = hashOps.multiGet(range);
        if (list != null) {
          List<SeckillSkuRedisTo> collect = list.stream().map(item -> {
            SeckillSkuRedisTo redisTo = JSON.parseObject((String) item, SeckillSkuRedisTo.class);
            //redisTo.setRandomCode(null); //当前秒杀开始了 需要随机码
            return redisTo;
          }).collect(Collectors.toList());
          return collect;
        }
        break;
      }
    }

    return null;
  }

  @Override
  public SeckillSkuRedisTo getSkuSeckillInfo(Long skuId) {
    //1、找到所有需要参与秒杀的商品的key信息
    BoundHashOperations<String, String, String> ops = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);

    Set<String> keys = ops.keys();
    if (keys != null & keys.size() > 0) {
      String regx = "\\d_" + skuId;
      for (String key : keys) {
        if (Pattern.matches(regx, key)) {
          String jsonStr = ops.get(key);
          SeckillSkuRedisTo redisTo = JSON.parseObject(jsonStr, SeckillSkuRedisTo.class);
          //随机码,要在秒杀时间到了才返回
          long current = new Date().getTime();
          if (current < redisTo.getStartTime() || current > redisTo.getEndTime()) {
            redisTo.setRandomCode(null);
          }
          return redisTo;
        }
      }
    }

    return null;
  }

  @Override
  public String kill(String killId, String key, Integer num) {
    long s1 = System.currentTimeMillis();
    MemberRespVo memberRespVo = LoginUserInterceptor.loginUser.get();
    BoundHashOperations<String, String, String> ops = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);

    String json = ops.get(killId);
    if (StringUtils.isEmpty(json)) {
      return null;
    } else {
      SeckillSkuRedisTo redisTo = JSON.parseObject(json, SeckillSkuRedisTo.class);
      //校验合法性
      long startTime = redisTo.getStartTime();
      long endTime = redisTo.getEndTime();
      long time = new Date().getTime();
      //校验时间的合法性
      if (time >= startTime && time <= endTime) {
        //2、校验随机码和商品id
        String randomCode = redisTo.getRandomCode();
        String sessionId_skuId = redisTo.getPromotionSessionId()+"_"+redisTo.getSkuId();
        if (randomCode.equals(key) && killId.equals(sessionId_skuId)) {
          //3、验证购物数量是否合理
          if (num <= redisTo.getSeckillLimit().intValue()) {
            //4、验证这个人是否已经购买过。  幂等性。  秒杀成功,就去占一个位,userId_sessionId_skuId
            String redisKey = memberRespVo.getId() + "_" + sessionId_skuId;
            Boolean b = redisTemplate.opsForValue().setIfAbsent(redisKey, num.toString(), endTime - startTime/*←自动过期*/, TimeUnit.MILLISECONDS);
            if(b){ //占位成功,说明还没买过
              RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + randomCode);
              boolean b1 = semaphore.tryAcquire(num); //一定要使用tryAcquire 它不会卡死, acquire会一直等待 直到获取到
              if (b1) { //秒杀成功
                //快速下单 发送MQ消息
                String timeId = IdWorker.getTimeId();
                SeckillOrderTo orderTo = new SeckillOrderTo();
                orderTo.setOrderSn(timeId);
                orderTo.setMemberId(memberRespVo.getId());
                orderTo.setNum(num);
                orderTo.setPromotionSessionId(redisTo.getPromotionSessionId());
                orderTo.setSkuId(redisTo.getSkuId());
                orderTo.setSeckillPrice(redisTo.getSeckillPrice());
                rabbitTemplate.convertAndSend("order-event-exchange","order.seckill.order",orderTo);
                long s2 = System.currentTimeMillis();
                log.info("耗时: {}", s2 - s1);
                return timeId;
              }

            }else{ //占位失败,说明已经买过了
              return null;
            }
          }
        }else{
          return null;
        }
      }else{
        return null;
      }
    }

    return null;
  }

  @Override
  public void uploadSeckillSkuLatest3Days() {
    //1、 扫描最近三天需要参与秒杀的活动
    R r = couponFeignService.getLatest3DaySession();
    if (r.getCode() == 0) {
      //上架商品
      List<SeckillSessionsWithSkus> data = r.getData(new TypeReference<List<SeckillSessionsWithSkus>>() {});
      if(data==null) return;

      //缓存到redis
      //1、缓存活动信息
      saveSessionInfos(data);
      //2、缓存活动的关联商品信息
      saveSessionSkuInfos(data);
    }
  }

  //缓存活动信息
  private void saveSessionInfos(List<SeckillSessionsWithSkus> sessions) {
    sessions.stream().forEach(session -> {
      Long startTime = session.getStartTime().getTime();
      Long endTime = session.getEndTime().getTime();
      String key = SESSIONS_CACHE_PREFIX + startTime + "_" + endTime;
      Boolean hasKey = redisTemplate.hasKey(key);
      if (!hasKey) {
        List<String> collect = session.getRelationSkus().stream().map(item -> item.getPromotionSessionId() + "_" + item.getSkuId().toString()).collect(Collectors.toList());
        redisTemplate.opsForList().leftPushAll(key, collect);
      }
    });
  }

  //缓存商品信息
  private void saveSessionSkuInfos(List<SeckillSessionsWithSkus> sessions) {
    sessions.stream().forEach(session -> {
      //准备hash操作
      BoundHashOperations<String, Object, Object> ops = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
      session.getRelationSkus().stream().forEach(seckillSkuVo -> {
        if (!ops.hasKey(seckillSkuVo.getPromotionSessionId().toString() + "_" + seckillSkuVo.getSkuId().toString())) {
          //缓存商品
          SeckillSkuRedisTo redisTo = new SeckillSkuRedisTo();
          //1、sku的基本数据
          R r = productFeignService.getSkuInfo(seckillSkuVo.getSkuId());
          if (r.getCode() == 0) {
            SkuInfoVo skuInfo = r.getData("skuInfo", new TypeReference<SkuInfoVo>() {
            });
            redisTo.setSkuInfoVo(skuInfo);
          }
          //2、sku的秒杀信息
          BeanUtils.copyProperties(seckillSkuVo, redisTo);
          //3、设置当前商品的的秒杀时间信息
          redisTo.setStartTime(session.getStartTime().getTime());
          redisTo.setEndTime(session.getEndTime().getTime());
          //4、商品的随机码
          String token = UUID.randomUUID().toString().replace("-", "");
          redisTo.setRandomCode(token);

          String s = JSON.toJSONString(redisTo);
          ops.put(seckillSkuVo.getPromotionSessionId().toString() + "_" + seckillSkuVo.getSkuId().toString(), s);

          //5、使用分布式信号量来做库存  --- 限流
          //商品可以秒杀的数量作为信号量
          //如果当前这个场次的商品的库存信息已经上架就不需要上架
          RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + token);
          semaphore.trySetPermits(seckillSkuVo.getSeckillCount().intValue());
        }
      });
    });
  }
}
